package com.lyz.redis.service;

import cn.hutool.core.collection.CollUtil;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;

public class StreamRedisService<V> extends RedisService<String,V> {

    public StreamRedisService(RedisTemplate<String, V> redisTemplate) {
        super(redisTemplate);
    }

    /**
     *
     * @param k
     * @param values
     * @return
     */
    public RecordId addMap(String k, Map<String,V> values){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.add(k, values);
    }

    /**
     *
     * @param record
     * @return
     */
    public RecordId addRecord(Record<String,V> record){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.add(record);
    }

    /**
     *
     * @param record
     * @return
     */
    public RecordId add(MapRecord<String,String,V> record){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.add(record);
    }

    /**
     * 对流进行修剪，限制长度
     * @param k
     * @param count
     * @return
     */
    public Long trim(String k,Long count){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.trim(k,count);
    }

    /**
     * 对流进行修剪，限制长度
     * @param k
     * @param count
     * @param approximateTrimming 近似修剪
     * @return
     */
    public Long trim(String k,Long count,Boolean approximateTrimming){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.trim(k,count,approximateTrimming);
    }

    /**
     * 删除消息
     * @param k
     * @param id
     * @return
     */
    public Long deleteByRecordIds(String k,RecordId... id){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.delete(k,id);
    }

    /**
     * 删除消息
     * @param record
     * @return
     */
    public Long deleteByRecord(Record<String,V> record){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.delete(record);
    }

    /**
     * 删除消息
     * @param k
     * @param recordIds
     * @return
     */
    public Long deleteByIds(String k,String... recordIds){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.delete(k, recordIds);
    }

    /**
     * 获取流包含的元素数量，即消息长度
     * @param k
     * @return
     */
    public Long len(String k){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.size(k);
    }

    /**
     *  获取消息列表，会自动过滤已经删除的消息
     * @param k
     * @param range
     * @return
     */
    public List<MapRecord<String, String,V>> range(String k, Range<String> range){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.range(k,range);
    }

    public List<MapRecord<String, String,V>> rangeAndLimit(String k, Range<String> range, RedisZSetCommands.Limit limit){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.range(k,range,limit);
    }

    public List<ObjectRecord<String,V>> rangeByClassType(Class<V> clazz, String k, Range<String> range){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.range(clazz,k,range);
    }

    public List<ObjectRecord<String,V>> rangeAndLimitByClassType(Class<V> clazz,String k, Range<String> range,RedisZSetCommands.Limit limit){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.range(clazz,k,range,limit);
    }
    public List<MapRecord<String, String,V>> reverseRange(String k, Range<String> range){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.reverseRange(k,range);
    }

    public List<MapRecord<String, String,V>> reverseRangeAndLimit(String k, Range<String> range, RedisZSetCommands.Limit limit){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.reverseRange(k,range,limit);
    }

    public List<ObjectRecord<String,V>> reverseRangeByClassType(Class<V> clazz, String k, Range<String> range){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.reverseRange(clazz,k,range);
    }

    public List<ObjectRecord<String,V>> reverseRangeAndLimitByClassType(Class<V> clazz,String k, Range<String> range,RedisZSetCommands.Limit limit){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.reverseRange(clazz,k,range,limit);
    }

    @SafeVarargs
    public final List<MapRecord<String, String,V>> read(StreamOffset<String>... offset){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.read(offset);
    }

    @SafeVarargs
    public final List<MapRecord<String, String,V>> read(Consumer consumer,StreamOffset<String>... offset){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.read(consumer,offset);
    }

    @SafeVarargs
    public final List<MapRecord<String, String,V>> read(StreamReadOptions options,Consumer consumer,StreamOffset<String>... offset){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.read(consumer, options,offset);
    }

    @SafeVarargs
    public final List<ObjectRecord<String,V>> read(Class<V> type,StreamOffset<String>... offset){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.read(type, offset);
    }

    @SafeVarargs
    public final List<ObjectRecord<String,V>> read(Class<V> type,Consumer consumer,StreamOffset<String>... offset){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.read(type,consumer,offset);
    }

    @SafeVarargs
    public final List<ObjectRecord<String,V>> reverseRange(Class<V> type,StreamReadOptions options,StreamOffset<String>... offset){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.read(type, options,offset);
    }

    @SafeVarargs
    public final List<ObjectRecord<String,V>> reverseRange(Class<V> type, StreamReadOptions options, Consumer consumer, StreamOffset<String>... offset){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.read(type,consumer, options,offset);
    }

    /**
     * 创建消费者组
     * @param k
     * @param group
     * @return
     */
    public String createGroup(String k,String group){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.createGroup(k,group);
    }

    public String create(String k,ReadOffset offset,String group){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.createGroup(k,offset, group);
    }

    /**
     * 查看流和消费者组的相关信息
     * @param k
     * @param group
     * @return
     */
    public StreamInfo.XInfoConsumers consumers(String k, String group){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.consumers(k,group);
    }

    public Long ack(String k, String group,String... recordIds){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.acknowledge(k,group,recordIds);
    }

    public Long ack(String k, String group,RecordId... recordIds){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.acknowledge(k,group,recordIds);
    }

    public Long ack(String k, Record<String,V> record){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        return stream.acknowledge(k,record);
    }

    /**
     * 消息是否确认
     * @param k 消息key
     * @param group 订阅组
     * @param consumer 消费者名称
     * @param recordIds 消息ID
     * @return boolean
     */
    public Boolean isAck(String k, String group,String consumer,RecordId... recordIds){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        PendingMessages pending = stream.pending(k, Consumer.from(group, consumer));
        if(recordIds.length<=1){
            return pending.isEmpty();
        }
        ArrayList<RecordId> ids = CollUtil.toList(recordIds);
        return ids.stream().anyMatch(b->CollUtil.contains(ids,b.getSequence()));
    }

    public Boolean groupExists(String k,String group){
        StreamOperations<String, String, V> stream = this.redisTemplate.opsForStream();
        StreamInfo.XInfoGroups groups = stream.groups(k);
        return !groups.isEmpty() && groups.stream().allMatch(g->Objects.equals(group,g.groupName()));
    }
}
